from pyspark.sql import SparkSession
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("MySQL to Spark SQL Analysis") \
    .config("spark.jars", "path_to_mysql_connector_jar/mysql-connector-java-8.0.26.jar") \
    .getOrCreate()

# MySQL 连接参数
mysql_url = "jdbc:mysql://localhost:3306/your_database"
mysql_properties = {
    "user": "your_username",
    "password": "your_password",
    "driver": "com.mysql.cj.jdbc.Driver"
}

# 读取 MySQL 数据库中的表
df = spark.read.jdbc(url=mysql_url, table="your_table", properties=mysql_properties)

# 显示前几行数据
df.show(5)

# 注册 DataFrame 为临时表
df.createOrReplaceTempView("your_table_view")

# 执行 SQL 查询进行统计分析
# 例如，计算每个年龄的平均收入
sql_query = """
SELECT age, AVG(income) AS avg_income
FROM your_table_view
GROUP BY age
ORDER BY age
"""
result_df = spark.sql(sql_query)

# 将 Spark DataFrame 转换为 Pandas DataFrame
pandas_result_df = result_df.toPandas()

# 输出结果
print(pandas_result_df)

# 可视化结果
plt.figure(figsize=(10, 6))
sns.barplot(data=pandas_result_df, x='age', y='avg_income')
plt.title('Average Income by Age')
plt.xlabel('Age')
plt.ylabel('Average Income')
plt.show()

# 关闭 SparkSession
spark.stop()